package app.dim;

import Bean.TableProcess;
import app.dwd.BaseAppV1;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import common.Constant;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import util.FlinkSinkUtil;
import util.JdbcUtil;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;

public class DimApp extends BaseAppV1 {
    public static void main(String[] args) {
        new DimApp().init(
                2002,
                2,
                "DimApp",
                Constant.TOPIC_ODS_DB
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //清洗数据
        SingleOutputStreamOperator<JSONObject> edStream = etl(stream);
        //cdc 读配置表数据
        SingleOutputStreamOperator<TableProcess> tpProcess = readTableProcess(env);
        //在phoenix中建表
        tpProcess= createDimTable(tpProcess);
        //数据流和配置的connect
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connectedSream = connect(edStream, tpProcess);
//        connectedSream.print();
        //删除不需要的字段列
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream = deleteNoNeedColumns(connectedSream);
        //把维度数据写出到phoenix中
        writeToPhoenix(resultStream);
    }

    private void writeToPhoenix(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getPhoeixSink());
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> deleteNoNeedColumns(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connectedStream) {
        return connectedStream
                .map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
                    @Override
                    public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> t) throws Exception {
                        JSONObject data = t.f0;
                        List<String> columns = Arrays.asList(t.f1.getSinkColumns().split(","));
                        data.keySet().removeIf(key -> !columns.contains(key)  && !"op_type".equals(key));
                        return t;
                    }
                });
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject,TableProcess>> connect(
            SingleOutputStreamOperator<JSONObject> dataStream,
            SingleOutputStreamOperator<TableProcess> confStream) {
        MapStateDescriptor<String, TableProcess> tpStateDesc = new MapStateDescriptor<String, TableProcess>("tpState", String.class,TableProcess.class);
        BroadcastStream<TableProcess> tpBcSteam = confStream.broadcast(tpStateDesc);
        return dataStream
                .connect(tpBcSteam)
                .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject, TableProcess>>() {
                    private HashMap<String, TableProcess> tpMap;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        Connection conn = JdbcUtil.getMysqlConection();
                        String sql = "select * from edu_config.table_process";
                        List<TableProcess> list = JdbcUtil.queryList(conn, sql, null, TableProcess.class);
//                        System.out.println("========================================="+list);
                        tpMap = new HashMap<>();
                        for (TableProcess tp : list) {
                            tpMap.put(tp.getSourceTable() + ":" + tp.getSourceType(), tp);
                        }
                    }

                    @Override
                    public void processElement(JSONObject obj , ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        String key = obj.getString("table") + ":ALL";
//                        System.out.println(key);
                        ReadOnlyBroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                        TableProcess tp  = state.get(key);
                        if (tp == null) {
                            tp = tpMap.get(key);
                        }
                        if (tp != null) {
                            JSONObject data = obj.getJSONObject("data");
                            data.put("op_type", obj.getString("type"));
                            out.collect(Tuple2.of(data,tp));
                        }
                    }

                    @Override
                    public void processBroadcastElement(TableProcess tp , Context ctx, Collector<Tuple2<JSONObject, TableProcess>> out) throws Exception {
                        BroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                        String key=tp.getSourceTable()+":"+tp.getSourceType();
                        state.put(key,tp);
                    }
                });
    }

    private SingleOutputStreamOperator<TableProcess> createDimTable(SingleOutputStreamOperator<TableProcess> tpProcess) {
        return tpProcess
                .filter(tp -> "dim".equals(tp.getSinkType()))
                .process(new ProcessFunction<TableProcess, TableProcess>() {
                    private Connection conn;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        conn= JdbcUtil.getPhoenixConnection();
                    }

                    @Override
                    public void close() throws Exception {
                        JdbcUtil.close(conn);
                    }

                    @Override
                    public void processElement(TableProcess tp, Context ctx, Collector<TableProcess> out) throws Exception {
                        String op = tp.getOp();
                        StringBuilder sql =null;
                        if ("r".equals(op) || "c".equals(op)) {
                            sql = getCreateTableSql(tp);
                        } else if ("d".equals(op)) {
                            sql = getDelTableSql(tp);
                        } else {
                            PreparedStatement ps = conn.prepareStatement(getDelTableSql(tp).toString());
                            ps.execute();
                            ps.close();

                            sql = getCreateTableSql(tp);
                        }
                        PreparedStatement ps = conn.prepareStatement(sql.toString());
                        ps.execute();
                        ps.close();
                        out.collect(tp);
                    }

                    private StringBuilder getDelTableSql(TableProcess tp ) {
                        System.out.println("维度删表语句: drop table "+ tp.getSinkTable());
                        return new StringBuilder("drop table " + tp.getSinkTable());
                    }

                    private StringBuilder getCreateTableSql(TableProcess tp) {
                        StringBuilder sql = new StringBuilder();
                        sql
                                .append("create table if not exists ")
                                .append(tp.getSinkTable())
                                .append("(")
                                .append(tp.getSinkColumns().replaceAll("[^,]+","$0 varchar"))
                                .append(", constraint pk primary key (")
                                .append(tp.getSinkPk() == null ? "id":tp.getSinkPk())
                                .append("))")
                                .append(tp.getSinkExtend() == null ? "" :tp.getSinkExtend());
                        System.out.println("维度建表语句: " + sql);
                        return sql;
                    }
                });
    }

    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env) {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("hadoop174")
                .port(3306)
                .databaseList("edu_config")
                .tableList("edu_config.table_process")
                .username("root")
                .password("aaaaaa")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();
        return env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(),"mysql-source")
                .map(new MapFunction<String, TableProcess>() {
                    @Override
                    public TableProcess map(String value) throws Exception {
                        JSONObject obj = JSON.parseObject(value);
                        String op = obj.getString("op");
                        TableProcess tp=null;
                        if ("d".equals(op)) {
                            tp = obj.getObject("before", TableProcess.class);
                        } else {
                            tp = obj.getObject("after", TableProcess.class);
                        }
                        tp.setOp(op);
                        return tp;
                    }

                });
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream
                .filter(new FilterFunction<String>() {
                    @Override
                    public boolean filter(String value) throws Exception {
                        try {
                            JSONObject obj = JSON.parseObject(value.replaceAll("bootstrap-", ""));
                            String type = obj.getString("type");
                            String data = obj.getString("data");
                            return "edu".equals(obj.getString("database"))
                                    && ("insert".equals(type) || "update".equals(type))
                                    && data != null
                                    && data.length() > 2;
                        } catch (Exception e) {
                            System.out.println("你的 json 格式数据异常: " + value);
                            return false;
                        }
                    }
                })
                .map(json -> JSON.parseObject(json.replaceAll("bootstrap-", "")));
    }
}
